package kk.learn.spark.work._4

import org.apache.spark.sql.SparkSession

/**
 * <p>
 *
 * </p>
 *
 * @author KK
 * @since 2021-03-05
 */

object Knn {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("Knn")
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("warn")

    // 读取文件

    //  读取Iris.csv文件
    val iris = spark.read.format("csv")
      .option("inferSchema", "true")
      .option("header", "true")
      .load("code/data/Iris.csv")

    // 载入为iris表
    iris
      .createOrReplaceTempView("iris")

    //  读取Iris_unkown.csv文件
    val unknown = spark.read.format("csv")
      .option("inferSchema", "true")
      .option("header", "true")
      .load("code/data/Iris_unkown.csv")

    // 载入为unknown表
    unknown
      .createOrReplaceTempView("unknown")


   // 将两个表进行cross join， 计算距离
    spark.sql(
      """
        |select
        |     unknown.id,
        |     sqrt(pow((iris.SepalLengthCm - unknown.SepalLengthCm), 2)
        |     + pow((iris.SepalWidthCm - unknown.SepalWidthCm), 2)
        |     + pow((iris.PetalLengthCm - unknown.PetalLengthCm), 2)
        |     + pow((iris.PetalWidthCm - unknown.PetalWidthCm), 2)) dist,
        |     iris.Species
        |from iris
        |cross join unknown
        |""".stripMargin).createOrReplaceTempView("unknown_dist")

    // 选取每个未知分类的记录的前9条记录
    // 对这9条记录的情况进行计数
    spark.sql(
      """
        |select
        |   id,
        |   sum(case Species when 'Iris-setosa' then 1 else 0 end) setosa_count,
        |   sum(case Species when 'Iris-versicolor' then 1 else 0 end) versicolor_count,
        |   sum(case Species when 'Iris-virginica' then 1 else 0 end) virginica_count
        |from
        |(
        |select
        |   id,
        |   dist,
        |   Species,
        |   row_number() over(partition by id order by dist) rank
        |from unknown_dist
        |) with_rank
        |where with_rank.rank <= 9
        |group by id
        |""".stripMargin).createOrReplaceTempView("unknown_stat")

    // 得出分析出来的分类情况
    spark.sql(
      """
        |select
        | id,
        | case
        |   when (setosa_count >= versicolor_count and setosa_count >= virginica_count)
        |   then 'Iris-setosa'
        |   when (versicolor_count > setosa_count and versicolor_count >= virginica_count)
        |   then 'Iris-versicolor'
        |   when (virginica_count > setosa_count and virginica_count > versicolor_count)
        |   then 'Iris-virginica'
        |   end Species
        |from unknown_stat
        |order by id
        |""".stripMargin).show
    spark.close()
  }

}
